--
--  Copyright 2022 The Open Islands Authors. All Rights Reserved.
--
--  Licensed under the Apache License, Version 2.0 (the "License");
--  you may not use this file except in compliance with the License.
--  You may obtain a copy of the License at
--
--      http://www.apache.org/licenses/LICENSE-2.0
--
--  Unless required by applicable law or agreed to in writing, software
--  distributed under the License is distributed on an "AS IS" BASIS,
--  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
--  See the License for the specific language governing permissions and
--  limitations under the License.
--
local _M = {
    _VERSION = '0.1'
}

local ngx = ngx
local route_table = require "route_table"
local utils = require "utils"
local apiserver = require "apiserver"
local router = require("router")
local config = require "config"
local constant = require("constant")
local tonumber = tonumber
local dns_utils = require "dns_utils"
local kafka = require("kafka")
local response_utils = require("response_utils")
local log = require("log")

function _M.routing()
    local request_headers = ngx.req.get_headers()
    local routing_info = utils.json_loads(request_headers["routing"])
    local task_id = routing_info["task-id"]
    local dest_node_id = routing_info["dest-node-id"]
    local dest_role = routing_info["dest-role"]
    local retries = routing_info["retries"]
    local msg = string.format("dest node id: %s, dest role: %s, federated id: %s", dest_node_id, dest_role, task_id)
    local address, tls_enabled, next_service, err
    if config.node_id ~= "0" and dest_node_id == tostring(config.node_id) then
        -- my party
        local body, err = _M.decompress(request_headers, true)
        if not body then
            response_utils.failed_response(101, err)
        end
        local res, err = kafka.publish(routing_info, request_headers, body)
        if res then
            response_utils.success_response({})
        else
            log.error(err)
            response_utils.failed_response(101, err)
        end
    else
        address, tls_enabled, next_service, err = route_table.get_next_route(dest_node_id, federated_packet_type)
        if address ~= nil and next_service == constant.FEDERATEDPROXY then
            -- only federatedproxy support compress data now
            _M.compress(request_headers)
        end
    end
    router.routing(address, tls_enabled, next_service, err)
end

function _M.compress(request_headers)
    if request_headers["request-content-encoding"] and #request_headers["request-content-encoding"] > 0 then
        return
    end
    if request_headers["encoding-request-content"] == "gzip" then
        log.debug("compress data")
        local level
        if request_headers["request-content-compress-Level"] then
            level = tonumber(request_headers["request-content-compress-level"])
        else
            level = config.federatedproxy["content_compress_level"]
        end
        local res, deflated, err = utils.compress(utils.read_request_body(), level)
        if res == true then
            ngx.req.set_body_data(deflated)
            ngx.req.clear_header("encoding-request-content")
            ngx.req.set_header("request-content-encoding", "gzip")
            log.debug("compress data done")
        end
    end
end


function _M.decompress(request_headers, return_data)
    if request_headers["request-content-encoding"] == "gzip" then
        log.debug("decompress data")
        local res, inflated, err = utils.decompress(utils.read_request_body())
        if res == true then
            log.debug("decompress data done")
            if not return_data then
                ngx.req.set_body_data(inflated)
                ngx.req.clear_header("request-content-encoding")
                return "", "success"
            else
                return inflated, "success"
            end
        else
            return nil, "decompress failed"
        end
    elseif return_data then
        return utils.read_request_body(), "success"
    else
        return "", "pass"
    end
end

_M.routing()
